package com.mtiiot.fsmanage.server;

import com.mtiiot.fsmanage.bean.common.ChartSeries;
import com.mtiiot.fsmanage.bean.common.SysTreeNode;
import com.mtiiot.fsmanage.bean.common.TableDataInfo;
import com.mtiiot.fsmanage.bean.metadata.mq.MqStatisticsData;
import com.mtiiot.fsmanage.bean.metadata.mq.MqStatisticsGroup;
import com.mtiiot.fsmanage.utils.Contants;
import com.mtiiot.fsmanage.utils.TimeUtil;
import com.mtiiot.fscore.utils.StringUtil;
import com.mtiiot.fsmanage.core.dao.BaseElasticsearch;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.stats.InternalStats;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Created by xyr on 2016/8/18.
 */

@Service
@Transactional
public class MqServer {

    @Autowired
    private BaseElasticsearch baseElasticsearch;
    private static String INDEX = Contants.INDEX_METADATA;
    private static String KAFKA_TYPE = Contants.TYPE_METADATA_KAFKAMQ;
    private static String REDIS_TYPE = Contants.TYPE_METADATA_REDISMQ;

    /**
     * 查询
     * */
    public Object mqQuery(String sysName,String l1Name,String l2Name,String l3Name,String mqType,String times,String queryKeyWord,String mqTopic) throws Exception {
        Client client=null;
        try{
            client=baseElasticsearch.getEsDataSource().getClient();
            BoolQueryBuilder bq= QueryBuilders.boolQuery();

            if(!StringUtil.isEmpty(sysName)){
                bq.must(QueryBuilders.termQuery("sysName", sysName));                 // Query
            }
            if(!StringUtil.isEmpty(l1Name)){
                bq.must(QueryBuilders.termQuery("l1Name", l1Name));                 // Query
            }
            if(!StringUtil.isEmpty(l2Name)){
                bq.must(QueryBuilders.termQuery("l2Name", l2Name));                 // Query
            }
            if(!StringUtil.isEmpty(l3Name)){
                bq.must(QueryBuilders.termQuery("l3Name", l3Name));                 // Query
            }
            if(!StringUtil.isEmpty(queryKeyWord)){
                bq.must(QueryBuilders.queryStringQuery(queryKeyWord));
            }
            // 根据队列的主题查询
            if (!StringUtil.isEmpty(mqTopic) && !"allTopic".equals(mqTopic)){
                bq.must(QueryBuilders.termQuery("mqTopic",mqTopic.toLowerCase()));
            }
            long[] l= TimeUtil.handleTimes(times);
            if(l!=null&&l.length>0){
                bq.must(QueryBuilders.rangeQuery("time").from(l[0]).to(l[1]));
            }
            QueryBuilder queryBuilder = bq;

            SearchRequestBuilder srb = client.prepareSearch(INDEX);
            if(!StringUtil.isEmpty(mqType)){
                if ("kafka".equals(mqType)){
                    srb.setTypes(KAFKA_TYPE);
                } else if ("redis".equals(mqType)){
                    srb.setTypes(REDIS_TYPE);
                } else if ("all".equals(mqType)){
                    srb.setTypes(KAFKA_TYPE,REDIS_TYPE);
                }
            }
            srb.addSort("time", SortOrder.DESC);  //按时间降序号
            srb.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
            srb.setFrom(0); // 设置查询数据的位置,分页用
            srb.setSize(500); // 设置查询结果集的最大条数
            srb.setExplain(true); // 设置是否按查询匹配度排序
            // 最后就是返回搜索响应信息
            srb.setQuery(bq);

            /*SearchResponse response =srb.execute().actionGet();
            SearchHits searchHits = response.getHits();
            SearchHit[] hits = searchHits.getHits();
            List<MqMetadata> list=new ArrayList<MqMetadata>();
            for (SearchHit searchHit : hits) {
                Map<String, Object> sm = searchHit.sourceAsMap();
                MqMetadata lm = new MqMetadata();
                lm.setSysId(sm.get("sysId").toString());
                lm.setSysName(sm.get("sysName").toString());
                lm.setL1Name(sm.get("l1Name").toString());
                lm.setL2Name(sm.get("l2Name").toString());
                lm.setL3Name(sm.get("l3Name").toString());
                lm.setYear(Integer.parseInt(sm.get("year").toString()));
                lm.setMonth(Integer.parseInt(sm.get("month").toString()));
                lm.setDay(Integer.parseInt(sm.get("day").toString()));
                long t=(Long)sm.get("time");
                lm.setTime(new Date(t));
                lm.setShowTime(sm.get("showTime").toString());
                lm.setMqTopic(sm.get("mqTopic").toString());
                lm.setMqKey(sm.get("mqKey").toString());
                lm.setMqValue(sm.get("mqValue").toString());
                lm.setHost(sm.get("host").toString());
                lm.setRequestType(sm.get("requestType").toString());
                lm.setRequestSize(Double.parseDouble(sm.get("requestSize").toString()));
                list.add(lm);
            }
            return list;*/

            TableDataInfo tdif=new TableDataInfo();
            SearchResponse response =srb.execute().actionGet();
            SearchHits searchHits = response.getHits();
            tdif.setCount(searchHits.totalHits());
            SearchHit[] hits = searchHits.getHits();

            List<Map<String, Object>> sms=new ArrayList<Map<String, Object>>();
            for (SearchHit searchHit : hits) {
                Map<String, Object> sm = searchHit.sourceAsMap();
                sms.add(sm);
            }
            tdif.setObject(sms);
            return tdif;
        }finally {
            if(client!=null){
                baseElasticsearch.getEsDataSource().releaseClient(client);
            }
        }
    }

    /**
     * 统计
     * */
    public Object mqStatistics(String sysName,String l1Name,String l2Name,String l3Name,String mqType,String times,String mqTopic) throws Exception {
        Client client=null;
        try{
            client=baseElasticsearch.getEsDataSource().getClient();
            BoolQueryBuilder bq=QueryBuilders.boolQuery();

            if(!StringUtil.isEmpty(sysName)){
                bq.must(QueryBuilders.termQuery("sysName", sysName));
            }
            if(!StringUtil.isEmpty(l1Name)){
                bq.must(QueryBuilders.termQuery("l1Name", l1Name));
            }
            if(!StringUtil.isEmpty(l2Name)){
                bq.must(QueryBuilders.termQuery("l2Name", l2Name));
            }
            if(!StringUtil.isEmpty(l3Name)){
                bq.must(QueryBuilders.termQuery("l3Name", l3Name));
            }
            long[] l= TimeUtil.handleTimes(times);
            if(l!=null&&l.length>0){
                bq.must(QueryBuilders.rangeQuery("time").from(l[0]).to(l[1]));
            }
            // 根据队列的主题查询
            if (!StringUtil.isEmpty(mqTopic) && !"allTopic".equals(mqTopic)){
                bq.must(QueryBuilders.termQuery("mqTopic",mqTopic.toLowerCase()));
            }
            QueryBuilder queryBuilder = bq;

            SearchRequestBuilder srb = client.prepareSearch(INDEX);
            if(!StringUtil.isEmpty(mqType)){
                if ("kafka".equals(mqType)){
                    srb.setTypes(KAFKA_TYPE);
                } else if ("redis".equals(mqType)){
                    srb.setTypes(REDIS_TYPE);
                } else if ("all".equals(mqType)){
                    srb.setTypes(KAFKA_TYPE,REDIS_TYPE);
                }
            }
            srb.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
            srb.setExplain(true); // 设置是否按查询匹配度排序
            // 最后就是返回搜索响应信息
            srb.setQuery(bq);

            String aggTremName="tremAgg";
            String aggStatsName="statsAgg";
            String statsType=null;
            if(StringUtil.isEmpty(sysName)){
                //统计系统级别的
                srb.addAggregation(AggregationBuilders.terms(aggTremName).field("sysName").size(500).subAggregation(AggregationBuilders.stats(aggStatsName).field("requestSize")));
                statsType="sysName";
            }else{
                if(StringUtil.isEmpty(l1Name)){
                    //统计l1级别的
                    srb.addAggregation(AggregationBuilders.terms(aggTremName).field("l1Name").size(500).subAggregation(AggregationBuilders.stats(aggStatsName).field("requestSize")));
                    statsType="l1";
                }else{
                    if(StringUtil.isEmpty(l2Name)){
                        //统计l2级别的
                        srb.addAggregation(AggregationBuilders.terms(aggTremName).field("l2Name").size(500).subAggregation(AggregationBuilders.stats(aggStatsName).field("requestSize")));
                        statsType="l2";
                    }else{
                        //统计l3级别的
                        srb.addAggregation(AggregationBuilders.terms(aggTremName).field("l3Name").size(500).subAggregation(AggregationBuilders.stats(aggStatsName).field("requestSize")));
                        statsType="l3";
                    }
                }
            }
            SearchResponse response =srb.execute().actionGet();
            Aggregations sysAggr=response.getAggregations();
            Terms sysNameAgg=sysAggr.get(aggTremName);
            List<Terms.Bucket> list=sysNameAgg.getBuckets();
            //LogStatisticsGroup lsg=new LogStatisticsGroup();
            MqStatisticsGroup lsg = new MqStatisticsGroup();
            //List<LogStatisticsData> lsds=new ArrayList<LogStatisticsData>();
            List<MqStatisticsData> lsds = new ArrayList<MqStatisticsData>();
            String[] chartCategories=new String[list.size()];
            Long[] requestCountArr=new Long[list.size()];
            Double[] requestSizeArr=new Double[list.size()];
            //LogStatisticsData lsd=null;
            MqStatisticsData lsd = null;
            int i=0;
            for(Terms.Bucket tb:list){
                Aggregations statsAggr=tb.getAggregations();
                InternalStats is = statsAggr.get(aggStatsName);
                String keyValue=tb.getKey().toString();
                chartCategories[i]=keyValue;
                requestCountArr[i]=is.getCount();
                requestSizeArr[i]=is.getSum();
                lsd = new MqStatisticsData();
                if("sysName".equals(statsType)){
                    lsd.setSysName(keyValue);
                }else if("l1".equals(statsType)){
                    lsd.setSysName(sysName);
                    lsd.setL1Name(keyValue);
                }else if("l2".equals(statsType)){
                    lsd.setSysName(sysName);
                    lsd.setL1Name(l1Name);
                    lsd.setL2Name(keyValue);
                }else if("l3".equals(statsType)){
                    lsd.setSysName(sysName);
                    lsd.setL1Name(l1Name);
                    lsd.setL2Name(l2Name);
                    lsd.setL3Name(keyValue);
                }
                lsd.setMqType(mqType);
                lsd.setMqTopic(mqTopic);
                lsd.setRequestCount(is.getCount());
                lsd.setRequestSize(is.getSum());
                lsds.add(lsd);
                i++;
            }
            lsg.setMqDataList(lsds);
            lsg.setChartCategories(chartCategories);
            List<ChartSeries> css=new ArrayList<ChartSeries>();
            if(requestCountArr.length>0){
                ChartSeries cs1=new ChartSeries();
                cs1.setName("请求");
                cs1.setData(requestCountArr);
                css.add(cs1);
                cs1=new ChartSeries();
                cs1.setName("流量");
                cs1.setData(requestSizeArr);
                css.add(cs1);
            }
            lsg.setCss(css);
            return lsg;
        }finally {
            if(client!=null){
                baseElasticsearch.getEsDataSource().releaseClient(client);
            }
        }
    }

    /**
     * 获得mq树
     * */
    public Object getMqChildrenNode(String id,String sysName,String l1Name,String l2Name,String kafkaType, String redisType) throws Exception {
        Client client=null;
        try{
            client=baseElasticsearch.getEsDataSource().getClient();
            SearchResponse sr=null;
            List<Terms.Bucket> list=null;
            List<SysTreeNode> sts=new ArrayList<SysTreeNode>();
            boolean hasChildren=true;
            if(id.equals("1")){
                //////////////////////查询系统////////////////
                sr = client.prepareSearch(INDEX)
                        .setTypes(kafkaType,redisType)
                        .setSize(0)
                        .setQuery(QueryBuilders.matchAllQuery())
                        .addAggregation(
                                AggregationBuilders.terms("agg1").field("sysName").size(500)
                        ).execute().actionGet();

            }else if(id.split("-").length==2){
                //////////////////////查询L1////////////////
                sr = client.prepareSearch(INDEX)
                        .setTypes(kafkaType,redisType)
                        .setSize(0)
                        .setQuery(QueryBuilders.matchAllQuery())
                        .setQuery(QueryBuilders.termQuery("sysName", sysName))                 // Query
                        .addAggregation(
                                AggregationBuilders.terms("agg1").field("l1Name").size(500)
                        ).execute().actionGet();
            }else if(id.split("-").length==3){
                //////////////////////查询L2////////////////
                sr = client.prepareSearch(INDEX)
                        .setTypes(kafkaType,redisType)
                        .setSize(0)
                        .setQuery(QueryBuilders.matchAllQuery())
                        .setQuery(QueryBuilders.termQuery("sysName", sysName))                 // Query
                        .setQuery(QueryBuilders.termQuery("l1Name", l1Name))                 // Query
                        .addAggregation(
                                AggregationBuilders.terms("agg1").field("l2Name").size(500)
                        ).execute().actionGet();
            }else if(id.split("-").length==4){
                //////////////////////查询L3////////////////1
                hasChildren=false;
                sr = client.prepareSearch(INDEX)
                        .setTypes(kafkaType,redisType)
                        .setSize(0)
                        .setQuery(QueryBuilders.matchAllQuery())
                        .setQuery(QueryBuilders.termQuery("sysName", sysName))                 // Query
                        .setQuery(QueryBuilders.termQuery("l1Name", l1Name))                 // Query
                        .setQuery(QueryBuilders.termQuery("l2Name", l2Name))                 // Query
                        .addAggregation(
                                AggregationBuilders.terms("agg1").field("l3Name").size(500)
                        ).execute().actionGet();
            }
            Terms agg1 = sr.getAggregations().get("agg1");
            list=agg1.getBuckets();
            int i=1;
            Object key=null;
            String text=null;
            for(Terms.Bucket b:list){
                SysTreeNode st=new SysTreeNode();
                st.setId(id+"-"+i);
                key=b.getKey();
                text=key.toString();
                st.setText(text);
                st.setChildren(hasChildren);
                sts.add(st);
                i++;
            }
            return sts;
        }finally {
            if(client!=null){
                baseElasticsearch.getEsDataSource().releaseClient(client);
            }
        }
    }

    /**
     * 获得队列的所有的主题
     * */
    public Object getAllTopic(String mqType) throws Exception {
        Client client = null;
        try {
            client = baseElasticsearch.getEsDataSource().getClient();
            SearchResponse sr = null;
            List<Terms.Bucket> list = null;
            List<SysTreeNode> sts = new ArrayList<SysTreeNode>();
            SearchRequestBuilder srb = client.prepareSearch(INDEX);
            if ("all".equals(mqType)){
                srb.setTypes(KAFKA_TYPE,REDIS_TYPE);
            } else if ("kafka".equals(mqType)){
                srb.setTypes(KAFKA_TYPE);
            } else if ("redis".equals(mqType)){
                srb.setTypes(REDIS_TYPE);
            }
            srb.setQuery(QueryBuilders.matchAllQuery());
            srb.addAggregation(AggregationBuilders.terms("agg1").field("mqTopic"));
            sr = srb.execute().actionGet();
            Terms agg1 = sr.getAggregations().get("agg1");
            list = agg1.getBuckets();
            int i = 1;
            Object key = null;
            String text = null;
            for (Terms.Bucket b : list) {
                SysTreeNode st = new SysTreeNode();
                //st.setId(id + "-" + i);
                key = b.getKey();
                text = key.toString().toUpperCase();
                st.setId(text);
                st.setText(text);
                sts.add(st);
                i++;
            }
            return sts;
        } finally {
            if(client!=null){
                baseElasticsearch.getEsDataSource().releaseClient(client);
            }
        }
    }

 }
